// Merger.java

package com.ibm.tspaces.examples.mergesort;


/*
**  Licensed Materials - Property of IBM
**
**  (C) COPYRIGHT IBM Corp. 1997  All rights reserved
**
**  US Government Users Restricted Rights - Use, duplication or
**  disclosure restricted by GSA ADP Schedule Contract with IBM Corp.
**
**/

import  java.util.Vector;
import  java.io.Serializable;
import  com.ibm.tspaces.*;
import  com.ibm.tspaces.services.utility.Eval;

/**
** This class implements a runnable object that when given control in the 
** run method, will start merging DataTuple objects that it finds in 
** the space.  It is designed to work with the Worker clients that are 
** similar in purpose to the Linda EVAL tuples.
**
** @see MergeSort
** @see Tuple
** @see Field
** @see TupleSpace
** @version $Revision: 2.1 $ $Date: 1999/11/05 22:22:39 $
** @author John Thomas
*/

public class Merger implements Eval {

  /*
  ***************************************************************************
  ** ** Members **
  ** *************
  **
  */

  /**
  **	The Host name where Tuples to be sorted will be found.
  */
  private String _tsHost = null;
  private String _tsName = null;
	
  
  private boolean _waitForT1 = false;
  /**
  **  The TupleSpace used to hold values.
  */
  private transient TupleSpace  ts  = null;

  private transient Thread runner = null;
	
	private Serializable _result = null;
  /*
  ***************************************************************************
  ** ** Merger **
  ** ****************
  */
  /**
  ** Create a Merger object.   
  ** The run method for this object will merge the mergesort data objects
  ** that it finds in the space.
  **
  ** @param host  the server where the TupleSpace resides
  ** @param name  the TupleSpace where the values to be sorted are stored.
  ** @see TupleSpace
  ***************************************************************************
  */
  public Merger(  String name, String host ) {
    
    this._tsHost = host;
    this._tsName = name;
  	
  } // Merger

  /*
  ***************************************************************************
  ** ** run **
  ** *********
  */
  /**
  ** The starting point of execution.
  ***************************************************************************
  */
  public void run()   {
    /**
    **  The TupleSpace used to hold values.
    */
    ts  = null;
    _result = new Integer(0);
    /** 
    ** Count of Tuple reads.
    */
    int  t1cnt = 0;
    int  t2cnt = 0;

    Debug.out( "Merger thread starting" );
    
    try {
    	ts = new TupleSpace(_tsName, _tsHost);
      // Make a tuple to be used as a template for retrieving tuples with
      // Vectors of integers to be merge sorted.
      DataTuple	formalDataTuple	=	new DataTuple();
      
      // Make a tuple to be used to as a tempalte for retrieving tuples
      // the count of the number of integers to sort.  This is used
      // to tell us when we are done sorting.
      CountTuple formalCountTuple = new CountTuple();
      
    
    	// First check if there are no data  tuples.  That means that 
    	// there is nothing left to sort and we can immediately return
    	int n = ts.countN(formalDataTuple);
    	if ( n == 0 ) {
    		Debug.out( "Merger thread has nothing to do");
    		return;
    	}
      // Loop forever
      while ( true ) {
        
        // Try to get a Vector of integers to be sorted from the TupleSpace
        // null will be returned if there is no such Vector
        Thread t = waitForTuple(ts,(SuperTuple)formalDataTuple,10);
        DataTuple  t1  =  (DataTuple)ts.waitToTake( formalDataTuple );
        if ( t1 == null) {
            Debug.out("Null returned from waitToTake");
            Debug.out("This should not happen. Reissue request");
            continue;
          }
      	Debug.out( "Merger  t1 = " + t1.toString() );
        if (t1.matches(formalDataTuple)) {
        	Debug.out("T1 timed out");
        	Debug.out(0,"T1Count = "+ t1cnt + "  T2Count = "+ t2cnt);
          break;
        }  	else 
      		t.interrupt();
        t1cnt++;
         // Got one Vector, try to get another so we can merge them.
        
      
        DataTuple t2  =  (DataTuple)ts.take( formalDataTuple );
        
        if ( t2 != null) {
          t2cnt++;
          Debug.out( "Merger: t2 = " + t2.toString() );
          ts.write( mergeTuples( t1, t2 ) );
        } else { // else didn't get second Vector
              // No
              // Didn't get the second vector.  This could be because there is
              // only one vector tuple in the space.  If this is the case then
              // we want to signal that the sorting is complete.  We do that by
              // getting the count of the number of integers to sort, if that
              // value equals the size of the Vector of integers then we are done.
              // If the value doesn't equal it then some other thread has another
              // vector.  The one that grabs the count first is the one responsible
              // for finishing off.  So if we didn't grab the count, we return the
              // the Vector, if we did get the count, then we wait for a second
              // vector to show up.
      
            Debug.out( "Merger: t2 = null, trying for the count" );
      
            CountTuple countTuple = (CountTuple)ts.take( formalCountTuple );
                // Did we get the count?
           if( countTuple != null ) {
              // Yes
              Debug.out( "Merger: count = " + countTuple.toString() );
      
              int count  = countTuple.getCount();
              
              VectorWithEquals vec  =  (VectorWithEquals)t1.getField(1).getValue();
              Debug.out( "Merger:  checking vector size " );
      
              // Is the count equal to the size of the Vector?
              if ( count == vec.size() ) {
                // Yes This means we're done.                 
                Debug.out(0,"Done! T1Count = "+ t1cnt + "  T2Count = "+ t2cnt);

                ts.write( new DoneTuple( vec ));
                break;
                // end if count == vecSize
              } else {
                // No
                // We're not done, but we have the count so somebody
                // else should be returning a "data" tuple to the
                // space, we'll wait for it.
                Debug.out( "Merger: size (" + vec.size() + " != " + 
                    count +") NOT equal wait for other datatuple " );
                do {
                  t2 = (DataTuple)ts.waitToTake( formalDataTuple );
                  if ( t2 == null) {
                    Debug.out("Null returned from waitToTake");
                    Debug.out("This should not happen.");
                    System.exit(1);
                  }
                } while (t2 == null);       
            
                Debug.out( "Merger: t2 = " + t2.toString() );
                // Merge the vectors and return the count.
                ts.write( mergeTuples( t1, t2 ) );
                ts.write( countTuple );
              
             } // else not
             // end if got the count
           } else {
              Debug.out( "Merger: didn't get count, returning t1, sleeping..." );
      
              // No
              // We didn't get the count so return the first Vector.
              ts.write( t1 );
      
              // And sleep a bit
              Thread.yield();  // Shouldn't sleep do this? DAF
              //pause( sleepTime );
      
           } // else didn't get the count
        } // else
      
      } // while True
	
    } // try
    catch ( Throwable t ) {
      Debug.out(t );
    	_result = t;    	
      
    } // catch

  } // run
	
/*
***************************************************************************
** ** getResult **
** ***************/
/**
** Returns the result of the run method
**
** @return Either the reusult of the merge or an exception
**
*/
public Serializable
getResult()          {
	
	return _result;
	
}
	
	
	
/*
***************************************************************************
** ** waitForTuple **
** *****************
*/

/**
** wait for a timeout interval and if not interrupted,
** then write a tuple that will satisfy the waitToTake
*/
private Thread 
waitForTuple(TupleSpace ts, SuperTuple satisfing, int timeout) {
  final TupleSpace tsF = ts;
	final int timeoutF = timeout;
	final SuperTuple satisfingF = satisfing;
  Thread inner = new Thread(new Runnable() {
    public void run() {
      try {
        Thread.sleep(timeoutF*1000);
      } catch (InterruptedException ie) {
        return;
      }
      try {
      	Debug.out("Timeout");
        tsF.write(satisfingF);
      } catch (TupleSpaceException tse) {
        Debug.out(tse);
      }
    }
  }, "satisfy");
	inner.start();
  return inner;
} // end waitForTuple()

  /*
  ***************************************************************************
  ** ** mergeTuples **
  ** *****************
  */
  /**
  ** Extract the two Vectors of integers from some data Tuples and merge sort
  ** them to produce a data Tuple with the values sorted.
  ** 
  ** @param t1 A Tuple with the second field a Vector of sorted integers.
  ** @param t2 A Tuple with the second field a Vector of sorted integers.
  ** @return A DataTuple that contains the sorted collection of the two 
  **           vectors extracted from the parameter Tuples.
  ** @exception TupleSpaceException if the format of the tuple is incorrect.
  ***************************************************************************
  */
  public DataTuple mergeTuples( DataTuple t1, DataTuple t2 ) throws TupleSpaceException {

	  return new DataTuple( mergeVectors( t1.getIntegers(), t2.getIntegers() ));

  } // mergeTuples

  /*
  ***************************************************************************
  ** ** mergeVectors **
  ** *****************
  */
  /**
  ** Merge sort two vectors of sorted integers.
  ** 
  ** @param v1 A Vector of integers in increasing sorted order.
  ** @param v2 A Vector of integers in increasing sorted order.
  ** @return A Vector of the integers merge sorted.
  ***************************************************************************
  */
  public VectorWithEquals mergeVectors( Vector v1, Vector  v2 ) {

	  int	v1Size	=	v1.size();
	  int	v2Size	=	v2.size();
	  int	cur1	=	0;
	  int	cur2	=	0;

	  VectorWithEquals	sorted	=	new VectorWithEquals( v1Size + v2Size );

	  int	limit	=	v1Size + v2Size;

	  for ( int i = 0; i < limit; i++ ) {
		  // Have we finished the first vector?
		  if ( cur1 < v1Size ) {
			  // No
			  // Have we finished the second vector?
			  if ( cur2 < v2Size ) {
				  // No
				  // Okay, find the integer with the smallest
				  // value and put it into the sorted vector
				  // Is v1 < v2?

				  if ( ((Integer)v1.elementAt( cur1 )).intValue() < 
						((Integer)v2.elementAt( cur2 )).intValue() ) {
					  // Yes
					  sorted.addElement( v1.elementAt( cur1++ ) );
				  } // if v1 < v2
				  else {
					  // No
					  sorted.addElement( v2.elementAt( cur2++ ) );
				  } // else v1 >= v2

			  } // if finished second
			  else {
				  // Yes
				  // Finished the second Vector, just tag on the first
				  for ( ; cur1 < v1Size; cur1++ ) {
					  sorted.addElement( v1.elementAt( cur1 ) );
				  } // for
				  i = limit;
			  } // else finished second

		  } // if finished first
		  else {
			  // Yes
			  // Finished the first Vector, just tag on the second.
			  for ( ; cur2 < v2Size; cur2++ ) {
				  sorted.addElement( v2.elementAt( cur2 ) );
			  } // for
			  i = limit;
		  } 
	  } // for

	  return sorted;

  } // mergeVectors

  /*
  ***************************************************************************
  ** ** puase **
  ** ***********
  */ 
  /**
  ** Suspend the thread for the indicated number of seconds.  This is useful
  ** for giving a user enough time to read a message displayed on the 
  ** console before the server goes down
  **
  ** @param seconds the number of seconds to pause
  **************************************************************************
  */
  private static void pause( int seconds ) {
    try {
		  Thread.sleep( seconds * 1000 );
    } // try
    catch( InterruptedException ie ) {}

  } // pause

  
} // Merger
/* $Log: Merger.java,v $
/* Revision 2.1  1999/11/05 22:22:39  estesp
/* Update revision number to 2.1 on all files
/*
/* Revision 1.1.1.1  1999/11/05 16:24:53  estesp
/* Imported 2.1.0 release into Austin CVS
/*
 * Revision 1.3  1999/09/13 05:10:37  jthomas
 * because Worker is in new package
 *
 * Revision 1.2  1999/06/17 05:39:39  thodes
 *   Updated the inital checkin to append the "Log" field and
 * convert CR/LF to CR (for win/unix interop).
 *
 *   The inital checkin lower-cased directrory names, and omitted
 * many dirs/files that will added later, e.g., MoDAL stuff.
 *
 * All of the above will need to be done to all VSS-to-CVS
 * converted sources.  In summary, in moving from VSS to CVS, do this:
 *  - lower-case dir names
 *  - CR/LF --> CR
 *  - append CVS "Log" entties
 *
 * (a script that helps with #2 and #3 is in
 *     tspaces1:~thodes/bin/update-keywords.sh)
 * 
 */

